package com.my.study.rocketmq.transaction;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.concurrent.TimeUnit;

/**
 * 生产者（事务消息）
 *
 * @author yyb8152
 * @since 20220509
 **/
public class Producer {
    public static void main(String[] args) throws Exception {
        //1.创建事务消息生产者producer，并制定生产者组名
        TransactionMQProducer producer = new TransactionMQProducer("TransactionGroup12");
        //2.指定Nameserver地址
        producer.setNamesrvAddr("39.101.165.196:9876");

        // 3、监听本地事务（执行本地事务的入口）
        producer.setTransactionListener(new TransactionListener() {
            /**
             * 在该方法中执行本地事务
             * @param message
             * @param arg
             * @return
             */
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object arg) {
                // 提交
                if (StringUtils.equals("TAGA", message.getTags())) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                    // 回滚
                } else if (StringUtils.equals("TAGB", message.getTags())) {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                    // 中间值
                } else if (StringUtils.equals("TAGC", message.getTags())) {
                    return LocalTransactionState.UNKNOW;
                }
                return LocalTransactionState.UNKNOW;
            }

            /**
             * 该方法是MQ进行消息事务状态回查的方法
             * @param messageExt
             * @return
             */
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                System.out.println("消息的Tag:" + messageExt.getTags());
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });

        //4.启动producer
        producer.start();

        String[] tags = {"TAGA", "TAGB", "TAGC"};
        for (int i = 0; i < 3; i++) {
            //4.创建消息对象，指定主题Topic、Tag和消息体
            /**
             * 参数一：消息主题Topic
             * 参数二：消息Tag
             * 参数三：消息内容
             */
            Message message = new Message("TransactionTopic", tags[i], ("Hello rocketMQ" + i).getBytes());

            /**
             * 5.发送事务消息
             * message :消息
             *
             * arg: null 针对整个消息进行事务控制，而不是某条消息控制
             */
            TransactionSendResult result = producer.sendMessageInTransaction(message, null);

            System.out.println("发送结果：" + result);
            //System.out.println("发送状态=" + sendStatus + "," + "消息id=" + msgId + "," + "消息队列id=" + messageQueueId);

            // 休眠1s
            TimeUnit.SECONDS.sleep(1);

        }

        //6.关闭生产者producer
        //producer.shutdown();
    }

}
